"""
Copyright (c) 2022 Huawei Technologies Co.,Ltd.

openGauss is licensed under Mulan PSL v2.
You can use this software according to the terms and conditions of the Mulan PSL v2.
You may obtain a copy of Mulan PSL v2 at:

          http://license.coscl.org.cn/MulanPSL2

THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
See the Mulan PSL v2 for more details.
"""
"""
Case Type   : 基础功能
Case Name   : create subscription copy_data为false时，refresh copy_data
Create At   : 2022/11/09
Owner       : opentestcase036
Description :
    1.两个集群均创建表1
    2.发布端集群在表1内插入数据
    3.创建发布端
    4.创建订阅端
      4.1创建订阅端设置copy_data值为空
      4.2创建订阅端设置copy_data值为‘false’
    5.查询订阅端表1数据
    6.两个集群均创建表2
    7.发布端在表1和表2内插入数据
    8.发布端添加表2,更新订阅端copy_data为true
    9.查询表1与表2数据
    10.两个集群均创建表3
    11.发布端插入数据
    12.发布端添加表3,更新订阅端copy_data为false
    13.查询表1,2,3内数据
    14.更新订阅端copy_data为true
    15.查询表3数据
    16.发布端在表3内插入数据
    17.查询表3数据
    18.refresh更新copy_data为非法值
    19.删除订阅
    20.删除发布
    21.清理环境
Expect      :
    1.成功
    2.成功
    3.成功
    4.部分测试点报错
      4.1合理报错
      4.2成功
    5.数据未同步
    6.成功
    7.成功
    8.成功
    9.表1同步1条数据，表2同步2条数据
    10.成功
    11.成功
    12.成功
    13.新数据在表1,2内同步,在表3内未同步
    14.成功
    15.表3内数据未同步
    16.成功
    17.新数据在表3内同步
    18.合理报错
    19.成功
    20.成功
    21.成功
History     :
"""
import unittest
import os

from yat.test import macro
from yat.test import Node

from testcase.utils.Logger import Logger
from testcase.utils.CommonSH import CommonSH
from testcase.utils.Common import Common
from testcase.utils.Constant import Constant

primary_sh = CommonSH('PrimaryDbUser')


@unittest.skipIf(3 != primary_sh.get_node_num(), '非1+2环境不执行')
class PubSubClass(unittest.TestCase):
    def setUp(self):
        self.log = Logger()
        self.log.info("-----------this is setup-----------")
        self.log.info(f"-----{os.path.basename(__file__)[:-3]} start-----")
        self.pri_user_pub = Node(node='PrimaryDbUser')
        self.pri_user_sub = Node(node='remote1_PrimaryDbUser')
        self.constant = Constant()
        self.com_sh_sub = CommonSH('remote1_PrimaryDbUser')
        self.com_pub = Common()
        self.tb_name1 = 'tb_pub_sub_case246_1'
        self.tb_name2 = 'tb_pub_sub_case246_2'
        self.tb_name3 = 'tb_pub_sub_case246_3'
        self.sub_name = "sub_case246"
        self.pub_name = "pub_case246"
        self.port = int(self.pri_user_pub.db_port) + 1
        self.wal_level = self.com_pub.show_param("wal_level")
        self.user_param_pub = f'-U {self.pri_user_pub.db_user} ' \
            f'-W {self.pri_user_pub.db_password}'
        self.user_param_sub = f'-U {self.pri_user_sub.db_user} ' \
            f'-W {self.pri_user_sub.db_password}'
        text = '----修改pg_hba expect:成功----'
        self.log.info(text)
        self.log.info('----发布端状态----')
        status = primary_sh.get_db_cluster_status('detail')
        self.log.info(status)
        self.assertEqual(status.count('Normal'), 4, '执行失败' + text)
        self.log.info('----订阅端状态----')
        status = \
            self.com_sh_sub.get_db_cluster_status(
                'detail', env_path=macro.DB_ENV_PATH_REMOTE1)
        self.log.info(status)
        self.assertEqual(status.count('Normal'), 4, '执行失败' + text)
        guc_res = primary_sh.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '',
            'all', False, False, '',
            f'host    replication  {self.pri_user_sub.db_user} '
            f'{self.pri_user_sub.db_host}/32 sha256')
        self.log.info(guc_res)
        self.assertTrue(guc_res, '执行失败:' + text)
        if 'logical' != self.wal_level:
            result = primary_sh.execute_gsguc(
                'set', self.constant.GSGUC_SUCCESS_MSG, 'wal_level=logical')
            self.assertTrue(result, '执行失败:' + text)
            result = primary_sh.restart_db_cluster(True)
            flg = self.constant.START_SUCCESS_MSG in result \
                  or 'Degrade' in result
            self.assertTrue(flg, '执行失败:' + text)
        guc_res = self.com_sh_sub.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '',
            'all', False, False, macro.DB_INSTANCE_PATH_REMOTE1,
            f'host    replication  {self.pri_user_pub.db_user} '
            f'{self.pri_user_pub.db_host}/32 sha256',
            macro.DB_ENV_PATH_REMOTE1)
        self.log.info(guc_res)
        self.assertTrue(guc_res, '执行失败:' + text)

    def test_pubsub(self):
        text = '--step1:两个集群均创建表1 expect:成功--'
        self.log.info(text)
        sql = f'''drop table if exists {self.tb_name1};
                create table {self.tb_name1}(i int,t text);'''
        result1 = primary_sh.execut_db_sql(
            sql, sql_type=self.user_param_pub)
        self.log.info(result1)
        self.assertIn(self.constant.TABLE_DROP_SUCCESS, result1,
                      '执行失败:' + text)
        self.assertEqual(result1.count(self.constant.TABLE_CREATE_SUCCESS),
                         1, '执行失败:' + text)
        result2 = self.com_sh_sub.execut_db_sql(sql, self.user_param_sub, None,
                                                macro.DB_ENV_PATH_REMOTE1)
        self.log.info(result2)
        self.assertIn(self.constant.TABLE_DROP_SUCCESS, result2,
                      '执行失败:' + text)
        self.assertEqual(result2.count(self.constant.TABLE_CREATE_SUCCESS),
                         1, '执行失败:' + text)

        text = '--step2:发布端在表1内插入数据 expect:成功--'
        self.log.info(text)
        sql = f"insert into {self.tb_name1} values(1,'first'),(2,'second');" \
              f"select pg_sleep(5.5);"
        result = primary_sh.execut_db_sql(sql,
                                          sql_type=self.user_param_pub)
        self.log.info(result)
        self.assertIn(self.constant.INSERT_SUCCESS_MSG, result, '执行失败' +
                      text)

        text = '--step3:创建发布端;expect:成功--'
        self.log.info(text)
        sql = f"drop publication if exists {self.pub_name};" \
              f"create publication {self.pub_name} " \
              f"for table {self.tb_name1};"
        result = primary_sh.execut_db_sql(sql,
                                          sql_type=self.user_param_pub)
        self.log.info(result)
        self.assertIn(self.constant.drop_pub_succ_msg, result,
                      '执行失败:' + text)
        self.assertIn(self.constant.create_pub_succ_msg, result,
                      '执行失败:' + text)
        self.assertNotIn(self.constant.SQL_WRONG_MSG[1], result,
                         '执行失败:' + text)

        text = '--step4:创建订阅端;expect:部分测试点合理报错--'
        self.log.info(text)
        result = self.com_sh_sub.execute_generate(
            macro.COMMON_PASSWD, env_path=macro.DB_ENV_PATH_REMOTE1)
        self.assertIn('', result, '执行失败:' + text)
        text = '--step4.1:创建订阅端,copy_data为空值;expect:失败--'
        sql1 = f"drop subscription if exists {self.sub_name};" \
               f"create subscription {self.sub_name} connection " \
               f"'host={self.pri_user_pub.db_host} " \
               f"port={self.port} " \
               f"user={self.pri_user_pub.db_user} " \
               f"dbname={self.pri_user_pub.db_name} " \
               f"password={self.pri_user_pub.ssh_password}' " \
               f"publication {self.pub_name} with (copy_data='');"
        result = self.com_sh_sub.execut_db_sql(sql1, self.user_param_sub, None,
                                               macro.DB_ENV_PATH_REMOTE1)
        self.log.info(result)
        self.assertIn(self.constant.drop_sub_succ_msg, result,
                      '执行失败:' + text)
        self.assertIn('ERROR', result, '执行失败:' + text)
        text = '--step4.2:创建订阅端,copy_data值为false;expect:成功--'
        sql2 = f"drop subscription if exists {self.sub_name};" \
               f"create subscription {self.sub_name} connection " \
               f"'host={self.pri_user_pub.db_host} " \
               f"port={self.port} " \
               f"user={self.pri_user_pub.db_user} " \
               f"dbname={self.pri_user_pub.db_name} " \
               f"password={self.pri_user_pub.ssh_password}' " \
               f"publication {self.pub_name} with (copy_data=false);"
        result = self.com_sh_sub.execut_db_sql(sql2, self.user_param_sub, None,
                                               macro.DB_ENV_PATH_REMOTE1)
        self.log.info(result)
        self.assertIn(self.constant.drop_sub_succ_msg, result,
                      '执行失败:' + text)
        self.assertIn(self.constant.create_sub_succ_msg,
                      result, '执行失败:' + text)

        text = '--step5:查询订阅端表1数据 expect:数据未同步--'
        self.log.info(text)
        sql = f"select * from {self.tb_name1};"
        result = self.com_sh_sub.execut_db_sql(sql, self.user_param_sub, None,
                                               macro.DB_ENV_PATH_REMOTE1)
        self.log.info(result)
        self.assertEqual(result.count('0 rows'), 1, '执行失败' + text)

        text = '--step6:两个集群均创建表2 expect:成功--'
        self.log.info(text)
        sql = f'''drop table if exists {self.tb_name2};
                create table {self.tb_name2}(i int,t text);'''
        result1 = primary_sh.execut_db_sql(
            sql, sql_type=self.user_param_pub)
        self.log.info(result1)
        self.assertIn(self.constant.TABLE_DROP_SUCCESS, result1,
                      '执行失败:' + text)
        self.assertEqual(result1.count(self.constant.TABLE_CREATE_SUCCESS),
                         1, '执行失败:' + text)
        result2 = self.com_sh_sub.execut_db_sql(sql, self.user_param_sub, None,
                                                macro.DB_ENV_PATH_REMOTE1)
        self.log.info(result2)
        self.assertIn(self.constant.TABLE_DROP_SUCCESS, result2,
                      '执行失败:' + text)
        self.assertEqual(result2.count(self.constant.TABLE_CREATE_SUCCESS),
                         1, '执行失败:' + text)

        text = '--step7:发布端插入数据 expect:成功--'
        self.log.info(text)
        sql = f"insert into {self.tb_name2} values(1,'first'),(2,'second');" \
              f"select pg_sleep(5.5);"
        result = primary_sh.execut_db_sql(sql,
                                          sql_type=self.user_param_pub)
        self.log.info(result)
        self.assertIn(self.constant.INSERT_SUCCESS_MSG, result, '执行失败' +
                      text)
        sql = f"insert into {self.tb_name1} values(3,'first');" \
              f"select pg_sleep(5.5);"
        result = primary_sh.execut_db_sql(sql,
                                          sql_type=self.user_param_pub)
        self.log.info(result)
        self.assertIn(self.constant.INSERT_SUCCESS_MSG, result, '执行失败' +
                      text)

        text = '--step8:发布端添加表2,更新订阅端copy_data为true expect:成功--'
        self.log.info(text)
        sql = f"alter publication {self.pub_name} add table {self.tb_name2};"
        result = primary_sh.execut_db_sql(sql,
                                          sql_type=self.user_param_pub)
        self.log.info(result)
        self.assertIn(self.constant.alter_pub_succ_msg, result, '执行失败' +
                      text)
        sql = f"alter subscription {self.sub_name} " \
              f"refresh publication with (copy_data=true);"\
              f"select pg_sleep(2);" \
              f"alter subscription {self.sub_name} " \
              f"refresh publication with (copy_data=true);" \
              f"select pg_sleep(2);"
        result = self.com_sh_sub.execut_db_sql(sql, self.user_param_sub, None,
                                               macro.DB_ENV_PATH_REMOTE1)
        self.log.info(result)
        self.assertIn(self.constant.alter_sub_succ_msg, result, '执行失败' +
                      text)

        text = '--step9:查询表1与表2数据 expect:表1同步1条数据,表2同步2条数据--'
        self.log.info(text)
        sql = f"select * from {self.tb_name1};"
        result = self.com_sh_sub.execut_db_sql(sql, self.user_param_sub, None,
                                               macro.DB_ENV_PATH_REMOTE1)
        self.log.info(result)
        self.assertEqual(result.count('1 row'), 1, '执行失败' + text)
        self.assertIn('3 | first', result, '执行失败' + text)
        sql = f"select * from {self.tb_name2};"
        result = self.com_sh_sub.execut_db_sql(sql, self.user_param_sub, None,
                                               macro.DB_ENV_PATH_REMOTE1)
        self.log.info(result)
        self.assertEqual(result.count('2 rows'), 1, '执行失败' + text)

        text = '--step10:两个集群均创建表3 expect:成功--'
        self.log.info(text)
        sql = f'''drop table if exists {self.tb_name3};
                  create table {self.tb_name3}(i int,t text);'''
        result1 = primary_sh.execut_db_sql(
            sql, sql_type=self.user_param_pub)
        self.log.info(result1)
        self.assertIn(self.constant.TABLE_DROP_SUCCESS, result1,
                      '执行失败:' + text)
        self.assertEqual(result1.count(self.constant.TABLE_CREATE_SUCCESS),
                         1, '执行失败:' + text)
        result2 = self.com_sh_sub.execut_db_sql(sql, self.user_param_sub, None,
                                                macro.DB_ENV_PATH_REMOTE1)
        self.log.info(result2)
        self.assertIn(self.constant.TABLE_DROP_SUCCESS, result2,
                      '执行失败:' + text)
        self.assertEqual(result2.count(self.constant.TABLE_CREATE_SUCCESS),
                         1, '执行失败:' + text)

        text = '--step11:发布端插入数据 expect:成功--'
        self.log.info(text)
        sql = f"insert into {self.tb_name3} values(1,'first'),(2,'second');" \
              f"select pg_sleep(5.5);" \
              f"insert into {self.tb_name1} values(4,'step11');" \
              f"select pg_sleep(5.5);" \
              f"insert into {self.tb_name2} values(4,'step11');"
        result = primary_sh.execut_db_sql(sql,
                                          sql_type=self.user_param_pub)
        self.log.info(result)
        self.assertIn(self.constant.INSERT_SUCCESS_MSG, result, '执行失败' +
                      text)

        text = '--step12:发布端添加表3,更新订阅端copy_data为false expect:成功--'
        self.log.info(text)
        sql = f"alter publication {self.pub_name} add table {self.tb_name3};"
        result = primary_sh.execut_db_sql(sql,
                                          sql_type=self.user_param_pub)
        self.log.info(result)
        self.assertIn(self.constant.alter_pub_succ_msg, result, '执行失败' +
                      text)
        sql = f"alter subscription {self.sub_name} " \
              f"refresh publication with (copy_data=false);" \
              f"select pg_sleep(2)" \
              f"alter subscription {self.sub_name}" \
              f"refresh publication with (copy_data=false);"
        result = self.com_sh_sub.execut_db_sql(sql, self.user_param_sub, None,
                                               macro.DB_ENV_PATH_REMOTE1)
        self.log.info(result)
        self.assertIn(self.constant.alter_sub_succ_msg, result, '执行失败' +
                      text)

        text = '--step13:查询表1,2,3内数据 expect:新数据在表1,2内同步,在表3内未同步--'
        self.log.info(text)
        sql = f"select * from {self.tb_name1};"
        result = self.com_sh_sub.execut_db_sql(sql, self.user_param_sub, None,
                                               macro.DB_ENV_PATH_REMOTE1)
        self.log.info(result)
        self.assertEqual(result.count('2 rows'), 1, '执行失败' + text)
        self.assertIn('4 | step11', result, '执行失败' + text)
        sql = f"select * from {self.tb_name2};"
        result = self.com_sh_sub.execut_db_sql(sql, self.user_param_sub, None,
                                               macro.DB_ENV_PATH_REMOTE1)
        self.log.info(result)
        self.assertEqual(result.count('3 rows'), 1, '执行失败' + text)
        self.assertIn('4 | step11', result, '执行失败' + text)
        sql = f"select * from {self.tb_name3};"
        result = self.com_sh_sub.execut_db_sql(sql, self.user_param_sub, None,
                                               macro.DB_ENV_PATH_REMOTE1)
        self.log.info(result)
        self.assertEqual(result.count('0 rows'), 1, '执行失败' + text)

        text = '--step14:更新订阅端copy_data为true expect:成功--'
        self.log.info(text)
        sql = f"alter subscription {self.sub_name} " \
              f"refresh publication with (copy_data=true);" \
              f"select pg_sleep(2)" \
              f"alter subscription {self.sub_name}" \
              f"refresh publication with (copy_data=true);"
        result = self.com_sh_sub.execut_db_sql(sql, self.user_param_sub, None,
                                               macro.DB_ENV_PATH_REMOTE1)
        self.log.info(result)
        self.assertIn(self.constant.alter_sub_succ_msg, result, '执行失败' +
                      text)

        text = '--step15:查询表3数据 expect:表3数据未同步--'
        self.log.info(text)
        sql = f"select * from {self.tb_name3};"
        result = self.com_sh_sub.execut_db_sql(sql, self.user_param_sub, None,
                                               macro.DB_ENV_PATH_REMOTE1)
        self.log.info(result)
        self.assertEqual(result.count('0 rows'), 1, '执行失败' + text)

        text = '--step16:发布端在表3内插入数据 expect:成功--'
        self.log.info(text)
        sql = f"insert into {self.tb_name3} values(5,'step16');" \
              f"select pg_sleep(5.5);"
        result = primary_sh.execut_db_sql(sql,
                                          sql_type=self.user_param_pub)
        self.log.info(result)
        self.assertIn(self.constant.INSERT_SUCCESS_MSG, result, '执行失败' +
                      text)

        text = '--step17:查询表3数据 expect:新数据已同步--'
        self.log.info(text)
        sql = f"select * from {self.tb_name3};"
        result = self.com_sh_sub.execut_db_sql(sql, self.user_param_sub, None,
                                               macro.DB_ENV_PATH_REMOTE1)
        self.log.info(result)
        self.assertEqual(result.count('1 row'), 1, '执行失败' + text)
        self.assertIn('5 | step16', result, '执行失败' + text)

        text = '--step18:refresh更新copy_data为非法值 expect:合理报错--'
        self.log.info(text)
        sql = f"alter subscription {self.sub_name} " \
              f"refresh publication with (copy_data=none);"
        result = self.com_sh_sub.execut_db_sql(sql, self.user_param_sub, None,
                                               macro.DB_ENV_PATH_REMOTE1)
        self.log.info(result)
        self.assertIn('ERROR', result, '执行失败' +
                      text)

    def tearDown(self):
        self.log.info('------------this is tearDown-------------')
        text = '--step19:删除订阅 expect:成功--'
        self.log.info(text)
        sql = f"drop subscription  {self.sub_name};"
        drop_sub_result = self.com_sh_sub.execut_db_sql(
            sql, self.user_param_sub, None, macro.DB_ENV_PATH_REMOTE1)
        self.log.info(drop_sub_result)

        text = '--step20:删除发布 expect:成功--'
        self.log.info(text)
        sql = f"drop publication {self.pub_name};"
        drop_pub_result = primary_sh.execut_db_sql(
            sql, sql_type=self.user_param_pub)
        self.log.info(drop_pub_result)

        text = '--step21:清理环境,删除表 expect:成功--'
        self.log.info(text)
        sql = f"drop table {self.tb_name1},{self.tb_name2},{self.tb_name3};"
        result1 = self.com_sh_sub.execut_db_sql(sql, self.user_param_sub, None,
                                                macro.DB_ENV_PATH_REMOTE1)
        self.log.info(result1)
        result2 = primary_sh.execut_db_sql(sql,
                                           sql_type=self.user_param_pub)
        self.log.info(result2)
        guc_res2 = primary_sh.execute_gsguc(
            'set', self.constant.GSGUC_SUCCESS_MSG,
            f'wal_level={self.wal_level}')
        self.log.info(guc_res2)
        guc_res1 = primary_sh.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '',
            'all', False, False, '',
            f'host    replication  {self.pri_user_sub.db_user} '
            f'{self.pri_user_sub.db_host}/32')
        self.log.info(guc_res1)
        guc_res = self.com_sh_sub.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '',
            'all', False, False, macro.DB_INSTANCE_PATH_REMOTE1,
            f'host    replication  {self.pri_user_pub.db_user} '
            f'{self.pri_user_pub.db_host}/32',
            macro.DB_ENV_PATH_REMOTE1)
        self.log.info(guc_res)
        result = primary_sh.restart_db_cluster(True)
        flg = self.constant.START_SUCCESS_MSG in result \
              or 'Degrade' in result
        self.assertTrue(flg, '执行失败:' + text)
        self.assertIn(self.constant.TABLE_DROP_SUCCESS, result1,
                      '执行失败:' + text)
        self.assertIn(self.constant.TABLE_DROP_SUCCESS, result2,
                      '执行失败:' + text)
        self.assertTrue(guc_res, '执行失败:' + text)
        self.assertTrue(guc_res1, '执行失败:' + text)
        self.assertTrue(guc_res2, '执行失败:' + text)
        self.assertIn(self.constant.drop_pub_succ_msg, drop_pub_result,
                      '执行失败' + text)
        self.assertIn(self.constant.drop_sub_succ_msg, drop_sub_result,
                      '执行失败' + text)
        self.log.info(f"-----{os.path.basename(__file__)[:-3]} end-----")